package chapter5.eg3;

import chapter5.AbstractConsumer;
import java.time.Duration;
import java.util.Collections;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class ConsumerRunnable extends AbstractConsumer implements Runnable {

  private final KafkaConsumer<String, String> consumer;

  public ConsumerRunnable(String groupId, String topic) {
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    this.consumer = new KafkaConsumer<>(properties);
    consumer.subscribe(Collections.singletonList(topic));
  }

  @Override
  public void run() {
    while (true) {
      ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
      for (ConsumerRecord<String, String> record : records) {
        System.out.println(Thread.currentThread().getName() + " consumed " + record.partition()
          + " th message with offset: " + record.offset());
      }
    }
  }
}
